
package cfar.ff.fun;

import cfar.actors.Node;
import cfar.ff.data.Assocs.Assoc2;
import cfar.ff.data.List;
import cfar.ff.data.Stream;
import cfar.ff.interfaces.Sequence;
import java.util.Iterator;

/**
 *
 * @author badi
 */
public class Seq {

    public static <S,A> A head(Sequence<S,A> seq){
        return seq.head();
    }

    public static <S,A> F1<Sequence<S,A>, A> head(){
        return new F1<Sequence<S, A>, A>() {

            public A f(Sequence<S, A> seq) {
                return Seq.head(seq);
            }
        };
    }

    public static <S,A> Sequence<S,A> tail(Sequence<S,A> seq){
        return seq.tail();
    }

    public static <S,A> int length(Sequence<S,A> seq){
        return seq.length();
    }

    public static <S,A> Sequence<S,A> cons(A a, Sequence<S,A> seq){
        return seq.cons(a);
    }

    public static <S,A> Sequence<S,A> snoc(A a, Sequence<S,A> seq){
        return seq.snoc(a);
    }

    public static <S,A> Sequence<S,A> reverse(Sequence<S,A> seq){
        Sequence<S,A> s = (Sequence<S, A>) seq.empty();
        for(A a : seq)
            s = cons(a, s);
        return s;
    }

    public static <S,A> Sequence<S,A> concat(Sequence<S,A> left, Sequence<S,A> right){
        Sequence<S,A> s = (Sequence<S, A>) left.empty();
        for(A a : left)
            s = snoc(a, s);
        for(A a : right)
            s = snoc(a, s);
        return s;
    }


    public static <S,A,B> Sequence<S, Assoc2<A,B>> zip(Sequence<S,A> as, Sequence<S,B> bs){
        return new Fun.zip().f(as, bs);
    }


    public static <S,A,B,C> Sequence<S, C> zipWith(F2<A,B,C> f, Sequence<S,A> as, Sequence<S,B> bs){
        return new Fun.zipWith().f(f, as, bs);
    }


    public static boolean and(Sequence<List,Boolean> seq){
        return new Fun.and().f(seq);
    }


    public static <S,A> Sequence<S,A> filter (F1<A, Boolean> f, Sequence<S, A> seq){
        Sequence<S,A> seq2 = (Sequence<S, A>) seq.<A>empty();
        for(A a : seq){
            if(f.f(a))
                seq2 = snoc(a, seq2);
        }
        return seq2;
    }


    public static <S,A,B> Sequence<S,B> map (F1<A,B> f, Sequence<S,A> seq){
        Sequence<S,B> seq2 = (Sequence<S, B>) seq.<B>empty();
        for(A a : seq)
            seq2 = snoc(f.f(a), seq2);
        return seq2;
    }


    public static <S,A> List<A> mkList (Sequence<S,A> seq){
        List<A> l = List.nil();
        for(A a : seq)
            l = l.snoc(a);
        return l;
    }

    public static <S,A> Sequence<S,A> mkSeq(Iterable<A> vals, Sequence<S,A> seq){
        Sequence<S,A> s = seq;
        for(A a : vals)
            s = s.snoc(a);
        return s;
    }


    public static <A> Iterator<A> cycle(final Iterable<A> seq){
        return new Iterator<A>() {

            private Iterator<A> itr = seq.iterator();

            public boolean hasNext() {
                return true;
            }

            public A next() {
                if (itr.hasNext())
                    return itr.next();
                else {
                    itr = seq.iterator();
                    return next();
                }
            }

            public void remove() {
                throw new UnsupportedOperationException("Not allowed.");
            }
        };
    }

    public static <S,A> Sequence<S,A> shuffle(Sequence<S,A> seq){
        return seq.shuffle();
    }


    public static <S,A,B> Sequence<S,B> parMap (F1<A,B> f, Sequence<S,A> seq){
        return new cfar.parallel.map.Map().f(f, seq);
    }

    public static <S,A,B> Sequence<S,B> parRMap ( F1<Sequence<Stream, B>, B> chooser
                           , Sequence<Stream, Integer>  redundancies
                           , F1<A, B>                   f
                           , final Sequence<S, A>       seq
                           )
    {
        return new cfar.parallel.map.RMap<S,A,B>().f(chooser, redundancies, f, seq);
    }

    public static <S,A> A parFoldl1 (F2<A,A,A> f, Sequence<S,A> seq){
        return (A) new cfar.parallel.reduce.foldl1().f(f, seq);
    }

    public static <S,A> A parRReduce( final Sequence<S, Integer>        redundancies
                                    , final F1<Sequence<S, A>, A>       chooser
                                    , final F2<A, A, A>                 f
                                    , final Sequence<S, A>              seq
                                    )
    {
        return new cfar.parallel.reduce.RReduce<S,A>().f(redundancies, chooser, f, seq);
    }

    public static <S,A> A parReduce( F2<A, A, A> f
                                   , Sequence<S, A> seq
                                   )
    {
        return new cfar.parallel.reduce.Reduce<S,A>().f(f, seq);
    }


    public static <S1,S2,A> A distRReduce( Sequence<S1, Integer> redundancies
                                         , F1<List<A>, A>        chooser
                                         , F0<Sequence<S2, Node>>get_nodes
                                         , F2<A, A, A>           f
                                         , Sequence<S1, A>       seq         )
    {
        return new cfar.distributed.reduce.RReduce<S1,S2,A>().f(redundancies, chooser, get_nodes, f, seq);
    }

    public static <S1,S2,A> A distReduce( F0<Sequence<S2, Node>> get_nodes
                                        , F2<A, A, A> reducer
                                        , Sequence<S1, A> seq
                                        )
    {
        return new cfar.distributed.reduce.Reduce<S1,S2,A>().f(get_nodes, reducer, seq);
    }


    public static <S1,S2,A,B> Sequence<S1,B> distMap ( Sequence<S2,Node> avail_nodes
                                                     , F1<A, B>          f
                                                     , Sequence<S1, A>   seq        )
    {
        return new cfar.distributed.map.Map<S1,S2,A,B>().f(avail_nodes, f, seq);
    }

    public static <S1,S2,A,B> Sequence<S1,B> distRMap( Sequence<S1, Integer> redundancies
                                                     , F1<List<B>, B>        chooser
                                                     , Sequence<S2, Node>    nodes
                                                     , F1<A, B>              f
                                                     , Sequence<S1, A>       seq         )
    {
        return new cfar.distributed.map.RMap<S1,S2,A,B>().f(redundancies, chooser, nodes, f, seq);
    }

}
